import com.nekonomics.lakehouse.common.FlinkRunner._
import com.nekonomics.lakehouse.common.util.SqlUtil._
import com.nekonomics.lakehouse.common.adt.PaymentStatus._


/**
 *
 * 业务过程为订单，
 * 商家粒度，
 * 度量值为订单数、商品数、销售金额、支付订单数、未支付订单数
 *  统计周期为天
 */

object OrderMerchantSummaryDailyDws {

  def main(args: Array[String]): Unit = {
    implicit val config: SqlRunnerConfig = SqlRunnerConfig(groupId = "order_merchant_summary_daily_dws", port = 8084, ckp = true)
    runSql { (_, tableEnv) =>
      tableEnv.executeSql(
        sql"""
             |CREATE TABLE IF NOT EXISTS dws.order_merchant_summary_daily_dws (
             |  merchant_id BIGINT,
             |  date_key DATE,             -- "2025-12-31"
             |  total_orders INT,
             |  total_items INT,
             |  total_sales DECIMAL(10,2),
             |  paid_orders INT,
             |  unpaid_orders INT,
             |  PRIMARY KEY (merchant_id, date_key) NOT ENFORCED
             |)
             |""".stripMargin
      )

      tableEnv.executeSql(
        sql"""
             |
             |INSERT INTO dws.order_merchant_summary_daily_dws
             |SELECT
             |  merchant_id,
             |  CAST(DATE_FORMAT(order_created_at, 'yyyy-MM-dd') AS DATE) AS date_key,
             |  CAST(COUNT(DISTINCT order_id) AS INT) AS total_orders,
             |  CAST(
             |    SUM(CASE WHEN payment_status = '$Success' THEN quantity ELSE 0 END) AS INT
             |    ) AS total_items,
             |  SUM(CASE WHEN payment_status = '$Success' THEN total_amount ELSE 0 END) AS total_sales,
             |  CAST(COUNT(DISTINCT CASE WHEN payment_status = '$Success' THEN order_id END) AS INT) AS paid_orders,
             |  CAST(COUNT(DISTINCT CASE WHEN payment_status <> '$Success' THEN order_id END) AS INT) AS unpaid_orders
             |FROM dwd.order_item_detail_dwd
             |GROUP BY
             |  merchant_id,
             |  DATE_FORMAT(order_created_at, 'yyyyMMdd'),
             |  DATE_FORMAT(order_created_at, 'yyyy-MM-dd');
             |""".stripMargin
      )
    }
  }








}
